const fs = require('fs')
const EventEmit = require('events')

class ReadStream extends EventEmit {
    constructor(path, options = {}) {
        super()
        this.path = path
        this.flags = options.flags || 'r';
        this.encoding = options.encoding || 'utf-8';
        this.start = options.start || 0
        this.end = options.end || undefined
        this.autoClose = options.autoClose || false
        this.highWaterMark = options.highWaterMark || 64 * 1024
        this.flowing = false // 判断暂停启用的状态
        this.offset = options.start // 每次读取的偏移量
        this.open()

        this.on('newListener', (type) => {
            if (type === 'data') { // 刚进来就直接绑定data  但是此时还没有fd
                this.flowing = true
                this.read()
            }
        })
    }
    destroy(err) {
        if (err) {
            this.emit('error', err)
        }
        if (this.autoClose) {
            this.emit('close')
        }
    }

    open() {
        fs.open(this.path, this.flags, (err, fd) => {
            if (err) {
                return this.destroy(err)
            }
            this.fd = fd
            this.emit('open', fd)
        })
    }
    read() {
        if (typeof this.fd !== 'number') {
            return this.once('open', () => this.read())
            // this.open是异步的  当没有fd时  注册一个 监听open的方法，重新调用read
        }
        const readLength = this.end ? Math.min(this.highWaterMark, this.end - this.offset + 1) : this.highWaterMark
        const buf = Buffer.alloc(readLength) // buffer 不能复用  buff是应用类型
        fs.read(this.fd, buf, 0, readLength, this.offset, (err, bytesRead) => {
            if (err) {
                return this.destroy(err)
            }
            if (bytesRead) {
                this.offset += bytesRead
                this.emit('data', buf.slice(0, bytesRead))
                if (this.flowing) {
                    this.read()
                }
            } else {
                this.emit('end')
                this.destroy()
            }


        })
    }

    pause() {
        this.flowing = false
    }

    resume() {
        if (!this.flowing) {
            this.flowing = true
            this.read()
        }
    }
    pipe(ws) {
        this.on('data', (chunk) => {
            let isDone = ws.write(chunk)
            if (!isDone) {
                this.pause()
            }
        })
        ws.on('drain', () => {
            this.resume()
        })
    }
}

module.exports = ReadStream